package cn.trigram.file;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import org.apache.commons.lang3.ArrayUtils;

public class FileSplitSegmentHandle {

  private final File sourceFile;

  private final byte[] separatorBytes;

  private final long perSegmentSize;


  /**
   * 处理时的ByteBuffer的容量,最大10MB
   */
  private final int capacity;


  public FileSplitSegmentHandle(File sourceFile, byte[] separatorBytes, long perSegmentSize,
      int bufferSize) {

    this.sourceFile = sourceFile;
    //    内容分割字节
    this.separatorBytes =
        ArrayUtils.isNotEmpty(separatorBytes) ? separatorBytes : System.lineSeparator().getBytes();
    this.perSegmentSize = perSegmentSize;
    this.capacity = Math.min(bufferSize, 10485760);
  }

  public void split(PerSegmentHandle segmentHandle) throws IOException {

    doSplit(segmentHandle);
  }

  private void doSplit(PerSegmentHandle segmentHandle) throws IOException {
//    创建只读文件通道
    try (
        RandomAccessFile randomAccessFile = new RandomAccessFile(this.sourceFile, "r");
        FileChannel channel = randomAccessFile.getChannel()
    ) {
      long totalSize = this.sourceFile.length();
      ByteBuffer buffer = ByteBuffer.allocate(this.capacity);
      byte[] bufferDest = new byte[capacity];

//    定义每个分割段的读取的起始、结束位置
      long previousPosition = 0L, afterPosition = totalSize, segmentNum = 1L;
//    读取位置小于总字节数才继续
      while (channel.position() < totalSize) {
//      粗略的以给定的参数值跨度
        channel.position(channel.position() + this.perSegmentSize);
//      确定精确分割的位置
        while (true) {
//        每次预读前清除缓存，并切换为写入模式
          buffer.clear();
          int readByteNum = channel.read(buffer);
          if (readByteNum == -1) {
//          读到结束还没有分割字节，直接将文件末尾赋值
            afterPosition = totalSize;
            break;
          }
//          切换为读取模式
          buffer.flip();
          buffer.get(bufferDest, 0, readByteNum);
//        寻找分割符的位置
          int indexOfSeparator = indexOfSeparator(bufferDest, readByteNum);
          if (indexOfSeparator != -1) {
//          找到后赋值当前分割段的结束位置
            afterPosition = channel.position() + indexOfSeparator;
            break;
          }
        }
        segmentHandle.exec(this.sourceFile.toPath().normalize().toString(), segmentNum,
            previousPosition, afterPosition);
//      将上个的结束位置赋值给当前段起始位置
        previousPosition = afterPosition;
        afterPosition = totalSize;
        segmentNum++;
      }
    }

  }

  private int indexOfSeparator(byte[] bytes, int endIndex) {

    int pos = -1;
    for (int i = 0; i < endIndex; i++) {
      if (bytes[i] != this.separatorBytes[0]) {
        continue;
      }
//      从bytes中与separator第一个字节相等的位置比较
      int flag = this.separatorBytes.length;
      for (int j = 0; j < this.separatorBytes.length; j++) {
        if (bytes[i + j] != this.separatorBytes[j]) {
          flag = j - 1;
          break;
        }
      }
//      相等表示找到，否则继续找
      if (flag == this.separatorBytes.length) {
        pos = i;
        break;
      }
    }
    return pos;
  }

  protected interface PerSegmentHandle {

    void exec(String sourceFilePath, long segmentNum, long startPosition,
        long endPosition);

  }

}
